/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import org.apache.log4j.Logger;
import org.zeromq.ZContext;
import org.zeromq.ZLoop;
import org.zeromq.ZMQ; 
import org.zeromq.ZMsg;
import org.zeromq.ZLoop.IZLoopHandler;
import org.zeromq.ZMQ.PollItem;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;

public class CSBClient {
	private String conn;
	private String id;
    private int timeout;
    private int retry;
    private ZContext ctx;
    private Socket client = null;
    
    private int retriesLeft;
	String tpc;
	String msg;

    private static Logger LOG = Logger.getLogger(CSBClient.class.getName());
    
    protected String getRid() {
    	return null;
    }
    
    public CSBClient(String conn, int timeout, int retry) {
        this.conn = conn;
        this.timeout = timeout;
        this.retry = retry;
    	ctx = new ZContext();

        conn();
    }

    private void conn() {
    	if (client != null)
    		ctx.destroySocket(client);
    	client = ctx.createSocket(ZMQ.DEALER);
        id = CSBUtils.sockId();
        client.setIdentity(id.getBytes());
        client.connect(conn);
        LOG.info(id + ": connected to router via " + conn + "...");
    }
    
    private IZLoopHandler proc = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
            ZMsg msg = ZMsg.recvMsg(client);
            if (msg != null) {
            	msg.unwrap();
            	String reply = new String(msg.pop().getData());
            	if (reply.equals(CSB.WORKER_ACK)) {
            		return -1;
            	} else {
            		LOG.warn(id + ": malformed reply from router: " + reply);
            	}                    
            	msg.destroy();
            } else {
            	LOG.error(id + ": receiving interrupted.");
            }
        	return 0;
        }
	};

	private IZLoopHandler resend = new IZLoopHandler () {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
        	if (--retriesLeft <= 0) {
                LOG.info(id + ": timeout, reconnecting...");
        		conn();
        		return -1;
        	}
        	
			LOG.error(id + ": no response from router, resending...");
			do_send();
            return 0;
        }
    };
    
    private void do_send() {
		String rid = getRid();
    	if (rid != null)
    		msg = "%" + rid + "%" + msg;
    	
    	ZMsg zmsg = ZMsg.newStringMsg(msg);
    	zmsg.push(tpc);
    	zmsg.send(client);
    }
    
    public void send(String topic, String message) {
    	tpc = topic;
    	msg = message;
    	do_send();
      
        retriesLeft = retry;
        
        ZLoop loop = new ZLoop();
		PollItem poller = new PollItem(client, Poller.POLLIN);
		loop.addPoller(poller, proc, null);
		loop.addTimer(timeout, retry, resend, null);        		
		loop.start();
		loop.destroy();
    }    
}

